const express = require('express');
const bodyParser = require('body-parser')
const http = require('http');
const app = express();
const { Worker } = require('worker_threads');
const os = require('os');
let workers = [];
function createModelWorker() {
  let cpul = Math.floor(os.cpus().length * 3 / 2)
  for (let i = 0; i < cpul; i++) {
    let w = new Worker('./worker.js');
    w.on('message', (msg) => {
      if (msg.type === 'online') {
        workers[i].online = true;
        workers[i].threadId = msg.threadId;
      } else if (msg.type === 'BM' || msg.type === 'RBM') {
        console.log('BM&&RBM')
        workers[i].ready = true;
      } else if (msg.type === 'FF') {
        console.log('FF')
        workers[i].lastOutput = msg.res;
        workers[i].lastOutputId = msg.id;
        workers[i].processing = false;
      } else if (msg.type === 'T') {
        console.log('T')
        workers[i].trainingId = msg.id;
        workers[i].processing = false;
        workers[i].trainingDoneTime = new Date().getTime();
      }
    })
    workers[i] = {
      w,
      ready: false,
      online: false,
      processing: false,
      trainingId: null,
      lastOutput: null,
      lastOutputId: null,
      threadId: null,
      trainingDoneTime:null
    };
  }
}
function postMessage(w, data, cb) {
  let fn = (msg) => {
    cb(msg);
  }
  w.once('message', fn);
  w.postMessage(data);
}
app.use((req, res, next) => {
  console.log(new Date(),req.url);
  next();
})
app.use(bodyParser.urlencoded({
  extended: true
}))
app.use(bodyParser.json());
app.get('/ml/status', (req, res) => {
  res.send({
    workers: workers.map(w => {
      return {
        ...w, w: undefined
      }
    })
  })
});
app.post('/ml/createModel', (req, res) => {
  let w = workers.find(wk => wk.threadId === Number(req.body.threadId))
  if (w) {
    if (!w.processing) {
      if (!w.ready) {
        postMessage(w.w, { type: 'BM', config: { layers: JSON.parse(req.body.layers), network: req.body.network } }, (msg) => {
          res.send({ status: 1, threadId: w.threadId });
        })
      } else {
        postMessage(w.w, { type: 'RBM', config: { layers: JSON.parse(req.body.layers), network: req.body.network } }, (msg) => {
          res.send({ status: 1, threadId: w.threadId });
        })
      }
    }
  } else {
    res.send({ status: 0, threadId: null });
  }
})
app.post('/ml/feedforward', (req, res) => {
  console.log(req.body);
  let w = null;
  for (let i of workers) {
    if (i.threadId === Number(req.body.threadId)) {
      w = i;
      break;
    }
  }
  if (!w.processing && w.ready) {
    w.processing = true;
    postMessage(w.w, { type: 'FF', data: JSON.parse(req.body.data), id: new Date().getTime() }, (msg) => {
      res.send({ status: 1, res: msg.res });
    })
  } else {
    res.send({ status: 0, res: null });
  }

})
app.post('/ml/train', (req, res) => {
  let w = null;
  for (let i of workers) {
    if (i.threadId === Number(req.body.threadId)) {
      w = i;
      break;
    }
  }
  if (!w.processing && w.ready) {
    res.send({ status: 1 });
    w.processing = true;
    w.trainingId = new Date().getTime();
    postMessage(w.w, { type: 'T', data: JSON.parse(req.body.data), lr: req.body.lr, trainingTime: req.body.time, id: new Date().getTime() }, () => { })
  } else {
    res.send({ status: 0 });
  }
})
app.use((req, res) => {
  res.status(404).end();
});
app.use((error, req, res, next) => {
  console.log(error);
  res.send('error');
})
createModelWorker();
http.createServer(app).listen(2500, () => {
  console.log('server start...at http://localhost:2500');
});